package stormdemo;

import org.apache.storm.Config;
import org.apache.storm.LocalCluster;
import org.apache.storm.topology.TopologyBuilder;
import org.apache.storm.tuple.Fields;

public class WorldCountMain {
	
	public static void main(String[] args) throws InterruptedException {
		
		TopologyBuilder builder = new TopologyBuilder();
		
		builder.setSpout("spout", new SentenceSpout());
		builder.setBolt("split", new SplitSentenceBlot()).shuffleGrouping("spout");
		builder.setBolt("wordcount", new WorldCountBolt()).fieldsGrouping("split", new Fields("world"));
		builder.setBolt("report", new ReportBolt()).globalGrouping("wordcount");
		
		Config config = new Config();
		
		LocalCluster cluster = new LocalCluster();
		cluster.submitTopology("wordcount", config, builder.createTopology());
		
		Thread.sleep(5000);
		cluster.killTopology("wordcount");
		cluster.shutdown();
	}
}
